CREATE TABLE jz_resumepost_etl_kafka (
	content string
)
with (
'connector' = 'kafka',
	'topic' = 'hdp_lbg_zhaopin_jz_app_resumepost_log',
	'properties.bootstrap.servers' = 'bh-kafka2-1.58dns.org:9092,bh-kafka2-2.58dns.org:9092,bh-kafka2-3.58dns.org:9092,bh-kafka2-4.58dns.org:9092,bh-kafka2-5.58dns.org:9092',
    'properties.group.id' = 'hdp_lbg_zhaopin_jz_app_resumepost_maidian',
	'properties.client.id' = 'hdp_lbg_zhaopin-hdp_lbg_zhaopin_jz_app_resumepost_log-v6Tjt',
	'scan.startup.mode' = 'latest-offset',
	'sink.parallelism' = '48',
	'format' = 'raw'
);

CREATE TABLE jz_app_resumepost_maidian (
  sort_id string comment '排序时间秒级',
  dt date comment '日期类型字段',
  appclient string comment 'AppActionContext，json格式',
  headers string comment 'headers',
  cookies string comment 'cookies',
  resumeid string comment '简历id',
  userid string comment '用户id',
  post_source string comment '发布来源',
  targetcateid string comment '期望职位类别',
  targetareaid string comment '期望工作城市',
  platform string comment '平台'
	)
with (
 'connector' = 'starrocks',
 'jdbc-url' = 'jdbc:mysql://shuserwh.ddb.58dns.org:33160/ddb58_zppushuserwh?useUnicode=true&characterEncoding=utf8&autoReconnect=true&failOverReadOnly=false',
 'load-url' = 'shuserwh-etl.ddb.58dns.org:33161',
 'database-name' = 'ddb58_zppushuserwh',
 'table-name' = 'jz_app_resumepost_maidian',
 'sink.buffer-flush.max-bytes'='94371840',
 'sink.buffer-flush.max-rows'='100000',
 'sink.buffer-flush.interval-ms'='30000',
 'sink.properties.column_separator' = '\x01',
 'sink.properties.row_delimiter' = '\x02',
 'username' = 'serwh_58dp',
 'password' = '494021345f2e688e'
)
;


insert into jz_app_resumepost_maidian
select
CAST(UNIX_TIMESTAMP() AS string) as sort_id
,CURRENT_DATE as dt
,appclient
,headers
,cookies
,resumeid
,userid
,post_source
,targetcateid
,targetareaid
,platform
from (
	select
	regexp_replace(if(splitindex(resumepost_log,'\t',1) is null,'-', splitindex(resumepost_log,'\t',1)),'"','') as appclient
	,regexp_replace(if(splitindex(resumepost_log,'\t',2) is null,'-', splitindex(resumepost_log,'\t',2)),'"','') as headers
	,regexp_replace(if(splitindex(resumepost_log,'\t',3) is null,'-', splitindex(resumepost_log,'\t',3)),'"','') as cookies
	,regexp_replace(if(splitindex(resumepost_log,'\t',4) is null,'-', splitindex(resumepost_log,'\t',4)),'"','') as resumeid
	,regexp_replace(if(splitindex(resumepost_log,'\t',5) is null,'-', splitindex(resumepost_log,'\t',5)),'"','') as userid
	,regexp_replace(if(splitindex(resumepost_log,'\t',6) is null,'-', splitindex(resumepost_log,'\t',6)),'"','') as post_source
	,regexp_replace(if(splitindex(resumepost_log,'\t',7) is null,'-', splitindex(resumepost_log,'\t',7)),'"','') as targetcateid
	,regexp_replace(if(splitindex(resumepost_log,'\t',8) is null,'-', splitindex(resumepost_log,'\t',8)),'"','') as targetareaid
	,regexp_extract(regexp_replace(splitindex(resumepost_log,'\t',1),'"',''),'appOs:(.*?),',1) as platform
from (
	select msg as resumepost_log from jz_resumepost_etl_kafka
)cc
)bb
;
